package com.bbx.flink.demo.ProcessFunctionAllDemo;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

@Slf4j
public class KeyedProcessFunctionDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSink print = env.socketTextStream("192.168.10.131", 10003)
                .map(i -> Integer.parseInt(i))
                .keyBy(i -> i % 2 == 0)
                .process(new MyKeyedProcessFunction())
                .print();

        env.execute();
    }
}

@Slf4j
class  MyKeyedProcessFunction extends KeyedProcessFunction<Boolean,Integer,Integer>{

    private static final long serialVersionUID = 1999918094878190559L;

    @Override
    public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
        log.info("value[{}]",value);
        Boolean currentKey = ctx.getCurrentKey();
        log.info("currentKey[{}]",currentKey);
        //定义侧输出
//        ctx.output();
        TimerService timerService = ctx.timerService();
        long currentProcessingTime = timerService.currentProcessingTime();
        log.info("currentProcessingTime[{}]",currentProcessingTime);
        //注册定时器，此处的时间时 绝对时间   从1970....
        timerService.registerProcessingTimeTimer(currentProcessingTime+5000L);
        //删除定时器
//        timerService.deleteEventTimeTimer(currentProcessingTime+2000L);


    }


    /**
     * 连续上升
     * 定时器调用方法
     * @param timestamp
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
        log.info("定时器调用方法[{}]",timestamp);
        Boolean currentKey = ctx.getCurrentKey();
        log.info("currentKey[{}]",currentKey);

    }
}


